package cn.z2huo.demo.kafka.apache.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * <p>手动提交偏移量的消费者，并且不依赖于消费者组，没有调用消费者的 subscribe 方法
 *
 * @author z2huo
 */
@Slf4j
public class ManualAssignmentPartitionConsumer {

    public static final String TOPIC = "test-topic";

    public static final String GROUP_ID = "my-consumer-group";

    public static void main(String[] args) {

    }

    /**
     * <p>不依赖消费者组，手动确认分区，并定位消费位置
     */
    public void consume() {
        Properties properties = ConsumerProperties.getConsumerProperties();
        properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        TopicPartition partition4 = new TopicPartition(TOPIC, 4);

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties)) {
            consumer.assign(Collections.singleton(partition4));

            long position;

            position = consumer.position(partition4);
            log.info("position is {}", position);

            consumer.seek(partition4, 5);

            position = consumer.position(partition4);
            log.info("position is {}", position);

//            OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(0);
//            consumer.seek(partition4, offsetAndMetadata);
//            consumer.seekToBeginning(Collections.singleton(partition4));
//            consumer.seekToEnd(Collections.singleton(partition4));

            // 不断地拉取和处理消息
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    log.info("Received message, partition is {}, offset is {}, key is {}, value is {}",
                            record.partition(), record.offset(), record.key(), record.value());
                }
            }
        }
    }

}
